import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.CategoryOrderStatisticsBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class DwsCategoryOrderStatisticsWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsCategoryOrderStatisticsWindow().start(
                14245,
                4,
                "dws_category_order_statistics_window",
                Constant.DWD_TRANS_ORDER
        );
    }

    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //kafkaStrDS.print();
        //4> {"id":"75188","order_id":"70813","user_id":"2452","province_id":"5","course_id":"439",
        // "course_name":"大数据技术之Zookeeper（2021最新版）","order_status":"1001","sessoin_id":"d955d1e6-fcac-497f-8958-06ebb0f0398d",
        // "sc":"2","origin_amount":"200.0","coupon_reduce":"0.0","final_amount":"200.0","ts":1731738018}
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(
                new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String jsonStr) throws Exception {
                        return JSON.parseObject(jsonStr);
                    }
                }
        );

        //以唯一主键分组去重
        KeyedStream<JSONObject, String> idKeyedDs = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));

        //定时器去重 只获取最后一个
        SingleOutputStreamOperator<JSONObject> distinctDs = idKeyedDs.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<JSONObject> jsonObjectValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {

                        ValueStateDescriptor<JSONObject> stateProperties =
                                new ValueStateDescriptor<JSONObject>("jsonObjectValueState", JSONObject.class);
                        jsonObjectValueState = getRuntimeContext().getState(stateProperties);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

                        JSONObject lastJsonObj = jsonObjectValueState.value();
                        if (lastJsonObj == null) {
                            jsonObjectValueState.update(jsonObj);

                            TimerService timerService = ctx.timerService();
                            long currentProcessingTime = timerService.currentProcessingTime();
                            timerService.registerProcessingTimeTimer(currentProcessingTime + 5000L);
                        }
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, JSONObject>.OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObj = jsonObjectValueState.value();
                        out.collect(jsonObj);
                    }
                }
        );

        KeyedStream<JSONObject, String> orderIdKeyedDs =
                distinctDs.keyBy(jsonObj -> jsonObj.getString("order_id"));

        SingleOutputStreamOperator<JSONObject> processDs = orderIdKeyedDs.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    ValueState<String> lastOrderIdState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor =
                                new ValueStateDescriptor<String>("lastOrderIdState", String.class);
                        lastOrderIdState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObj, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        JSONObject jsonObject = new JSONObject();
                        Long ts = jsonObj.getLong("ts") * 1000;
                        String amount = jsonObj.getString("final_amount");
                        String courseId = jsonObj.getString("course_id");
                        String curOrderId = jsonObj.getString("order_id");
                        //上一个orderid，因为以orderId分组，如果第二次orderid等于上一次orderid说明是同一个订单
                        String lastOrderId = lastOrderIdState.value();

                        jsonObject.put("course_id",courseId);
                        jsonObject.put("order_user", 1);
                        jsonObject.put("order_count", 1);
                        jsonObject.put("amount", amount);
                        jsonObject.put("ts", ts);

                        if (lastOrderId != null && lastOrderId.equals(curOrderId)) {
                            jsonObject.put("order_user", 0);
                            jsonObject.put("order_count", 0);
                        }

                        lastOrderIdState.update(curOrderId);
                        out.collect(jsonObject);

                    }
                }
        );

        //processDs.print();
        //4> {"course_id":"60","amount":"200.0","order_count":1,"order_user":1,"ts":1731738631000}
        //4> {"course_id":"378","amount":"200.0","order_count":1,"order_user":1,"ts":1731738631000}
        SingleOutputStreamOperator<CategoryOrderStatisticsBean> beanDs = processDs.map(
                new MapFunction<JSONObject, CategoryOrderStatisticsBean>() {
                    @Override
                    public CategoryOrderStatisticsBean map(JSONObject value) throws Exception {
                        CategoryOrderStatisticsBean bean = CategoryOrderStatisticsBean.builder()
                                .stt("")
                                .edt("")
                                .curDate("")
                                .category("")
                                .subject("")
                                .course(value.getString("course_id"))
                                .amount(value.getDouble("amount"))
                                .orderUser(value.getLong("order_user"))
                                .orderCount(value.getLong("order_count"))
                                .ts(value.getLong("ts"))
                                .build();
                        return bean;
                    }
                }
        );

        //beanDs.print();
        SingleOutputStreamOperator<CategoryOrderStatisticsBean> withWatermarkDs = beanDs.assignTimestampsAndWatermarks(
                WatermarkStrategy.<CategoryOrderStatisticsBean>forMonotonousTimestamps()
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<CategoryOrderStatisticsBean>() {
                                    @Override
                                    public long extractTimestamp(CategoryOrderStatisticsBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )

        );

        KeyedStream<CategoryOrderStatisticsBean, String> CourseKeyedDs = withWatermarkDs.keyBy(CategoryOrderStatisticsBean::getCourse);

        WindowedStream<CategoryOrderStatisticsBean, String, TimeWindow> windowDs = CourseKeyedDs.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<CategoryOrderStatisticsBean> reduceDs = windowDs.reduce(
                new ReduceFunction<CategoryOrderStatisticsBean>() {
                    @Override
                    public CategoryOrderStatisticsBean reduce(CategoryOrderStatisticsBean value1, CategoryOrderStatisticsBean value2) throws Exception {
                        value1.setAmount(value1.getAmount() + value2.getAmount());
                        value1.setOrderUser(value1.getOrderUser() + value2.getOrderUser());
                        value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                        return value1;
                    }
                },
                new WindowFunction<CategoryOrderStatisticsBean, CategoryOrderStatisticsBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<CategoryOrderStatisticsBean> input, Collector<CategoryOrderStatisticsBean> out) throws Exception {
                        CategoryOrderStatisticsBean categoryOrderStatisticsBean = input.iterator().next();
                        CategoryOrderStatisticsBean bean = CategoryOrderStatisticsBean.builder()
                                .stt(DateFormatUtil.tsToDateTime(window.getStart()))
                                .edt(DateFormatUtil.tsToDateTime(window.getEnd()))
                                .curDate(DateFormatUtil.tsToDate(window.getStart()))
                                .category("")
                                .subject("")
                                .course(categoryOrderStatisticsBean.getCourse())
                                .amount(categoryOrderStatisticsBean.getAmount())
                                .orderUser(categoryOrderStatisticsBean.getOrderUser())
                                .orderCount(categoryOrderStatisticsBean.getOrderCount())
                                .ts(categoryOrderStatisticsBean.getTs())
                                .build();
                        out.collect(bean);
                    }
                }
        );

        //reduceDs.print();

        SingleOutputStreamOperator<CategoryOrderStatisticsBean> withCourseDs = AsyncDataStream.unorderedWait(
                reduceDs,
                new DimAsyncFunction<CategoryOrderStatisticsBean>() {
                    @Override
                    protected void addDim(CategoryOrderStatisticsBean obj, JSONObject dimJsonObj) {
                        obj.setCourse(dimJsonObj.getString("course_name"));
                        obj.setSubject(dimJsonObj.getString("subject_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_course_info";
                    }

                    @Override
                    public String getRowKey(CategoryOrderStatisticsBean obj) {
                        return obj.getCourse();
                    }
                }, 60, TimeUnit.SECONDS
        );

        SingleOutputStreamOperator<CategoryOrderStatisticsBean> withSubjectDs = AsyncDataStream.unorderedWait(
                withCourseDs,
                new DimAsyncFunction<CategoryOrderStatisticsBean>() {
                    @Override
                    protected void addDim(CategoryOrderStatisticsBean obj, JSONObject dimJsonObj) {
                        obj.setSubject(dimJsonObj.getString("subject_name"));
                        obj.setCategory(dimJsonObj.getString("category_id"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_subject_info";
                    }

                    @Override
                    public String getRowKey(CategoryOrderStatisticsBean obj) {
                        return obj.getSubject();
                    }
                }, 60, TimeUnit.SECONDS
        );

        SingleOutputStreamOperator<CategoryOrderStatisticsBean> withCategoryDs = AsyncDataStream.unorderedWait(
                withSubjectDs,
                new DimAsyncFunction<CategoryOrderStatisticsBean>() {
                    @Override
                    protected void addDim(CategoryOrderStatisticsBean obj, JSONObject dimJsonObj) {
                        obj.setCategory(dimJsonObj.getString("category_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_base_category_info";
                    }

                    @Override
                    public String getRowKey(CategoryOrderStatisticsBean obj) {
                        return obj.getCategory();
                    }
                }, 60, TimeUnit.SECONDS
        );

        //withCategoryDs.print();
        //4> CategoryOrderStatisticsBean(stt=2024-11-16 15:07:50, edt=2024-11-16 15:08:00, curDate=2024-11-16,
        // category=编程技术, subject=前端学科, course=2021 React全家桶, amount=600.0, orderUser=4, orderCount=3, ts=1731740874000)
        SingleOutputStreamOperator<String> jsonBean = withCategoryDs.map(new BeanToJsonStrMapFunction<>());
        jsonBean.print();
        //{"stt":"2024-11-16 15:11:00","edt":"2024-11-16 15:11:10","cur_date":"2024-11-16","category":"编程技术",
        // "subject":"j2ee学科","course":"Spring4.0","amount":200.0,"order_user":1,"order_count":1}
        jsonBean.sinkTo(FlinkSinkUtil.getDorisSink("dws_category_order_statistics_window"));
    }
}
